package com.gitee.xhs.savePointReader.broadcastState;

import com.gitee.xhs.EnvironmentConfig;
import com.gitee.xhs.FlinkEnvironmentFactory;
import com.gitee.xhs.broadcastState.BroadcastStateDao;
import com.gitee.xhs.util.FileUtil;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.state.api.SavepointReader;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Map;

import static com.gitee.xhs.StateConstant.CHECK_POINT_BASE_PATH;
import static com.gitee.xhs.StateConstant.CHECK_POINT_PREFIX;
import static com.gitee.xhs.broadcastState.BroadcastStateConstant.*;
import static com.gitee.xhs.saveAndModifyingPointWriter.SavePointWriterConstant.MODIFYING_UID_PREFIX;
import static com.gitee.xhs.util.SavePointApiUtil.MODIFYING_NAME;

/**
 * @author BigHead
 * @date 2022/7/7 10:55
 * @description BroadcastStateReaderMain
 */
public class BroadcastStateReaderMain {

    public static void main(String[] args) throws Exception {
        Map<String, String> pathMap = FileUtil.listHasStateFilePath(CHECK_POINT_BASE_PATH + CUSTOM_BROADCAST_STATE_DIR);

        StreamExecutionEnvironment streamExecutionEnvironment = FlinkEnvironmentFactory.simpleEnv(new EnvironmentConfig(null));
        for (Map.Entry<String, String> pathMapEntry : pathMap.entrySet()) {

            SavepointReader savepointReader = SavepointReader.read(
                    streamExecutionEnvironment,
                    CHECK_POINT_PREFIX + pathMapEntry.getValue(),
                    new HashMapStateBackend()
            );
            
            savepointReader
                    .readBroadcastState(
                            BROADCAST_STATE_UID,
                            BROADCAST_STATE_NAME,
                            TypeInformation.of(Long.class),
                            TypeInformation.of(BroadcastStateDao.class)
                    )
                    .map(
                            new BroadcastStateReaderMapFunction(pathMapEntry.getKey(), BROADCAST_STATE_UID)
                    )
                    .print();

            if (pathMapEntry.getKey().startsWith(MODIFYING_NAME)){
                savepointReader
                        .readBroadcastState(
                                BROADCAST_STATE_UID + MODIFYING_UID_PREFIX,
                                BROADCAST_STATE_NAME,
                                TypeInformation.of(Long.class),
                                TypeInformation.of(BroadcastStateDao.class)
                        )
                        .map(
                                new BroadcastStateReaderMapFunction(pathMapEntry.getKey(), BROADCAST_STATE_UID + MODIFYING_UID_PREFIX)
                        )
                        .print();
            }
            
        }
        streamExecutionEnvironment.execute("BroadcastStateReaderMain");

    }
    
}
